package hyl.base.mq.active;

import java.util.Enumeration;
import java.util.HashMap;
import java.util.Map;
import javax.jms.BytesMessage;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnectionFactory;
import hyl.core.run.IDo;

/**
 * 定位只传输小数据.快速接收
 * 
 * 
 */
public class MyActiveMq {
	/**
	 * 事务管理 AUTO_ACKNOWLEDGE = 1 ：自动确认 CLIENT_ACKNOWLEDGE = 2：客户端手动确认
	 * DUPS_OK_ACKNOWLEDGE = 3： 自动批量确认 SESSION_TRANSACTED = 0：事务提交并确认
	 * 但是在activemq补充了一个自定义的ACK模式: INDIVIDUAL_ACKNOWLEDGE = 4：单条消息确认
	 */
	Session _session;
	// 每个线程 都有一个独立的MessageProducer
	// ThreadLocal<MessageProducer> LocalProducer = new ThreadLocal<>();
	// ThreadLocal<MessageConsumer> LocalConsumer = new ThreadLocal<>();
	MessageProducer producer;
	MessageConsumer consumer;
	Destination _destination;
	// queue模式 可以串行测试
	// topic模式只能多线程测试
	public static final String TYPE_QUEUE = "queue";
	public static final String TYPE_TOPIC = "topic";
	private static final String TOPIC_DEFAULT = "topic.default";
	// 链接工厂
	// ConnectionFactory connectionFactory;
	// private ActiveMQConnectionFactory connectionFactory;
	// 链接对象
	private Connection connection;
	public static final int QUEUE_MODEL = 1;
	public static final int TOPIC_MODEL = 2;
	private int _acknowledgeMode = Session.AUTO_ACKNOWLEDGE;

	// 检查Activemq是否可用
	public static boolean checkConnectConfig(ActConfig aconfig) {
		ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(aconfig.getUrl());
		// factory.setCacheDestinations(true);
		try {
			// "admin", "activemq"
			Connection connection = factory.createConnection();
			connection.start();
			Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
			// Create the destination (Topic or Queue)
			Destination destination = session.createTopic(TOPIC_DEFAULT);
			// Create a MessageProducer from the Session to the Topic or Queue
			MessageProducer producer = session.createProducer(destination);
			// producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
			// Create a messages
			String text = "Hello world! From: " + Thread.currentThread().getName();
			TextMessage message = session.createTextMessage(text);
			// Tell the producer to send the message
			System.out.println(text);
			producer.send(message);
			MessageConsumer cons = session.createConsumer(destination);
			TextMessage message1 = (TextMessage) cons.receive();
			System.out.println("ww=" + message1.getText());
			// Clean up
			session.close();
			connection.close();
			return true;
		} catch (JMSException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			return false;
		}
	}

	void connect(ActConfig aconfig) throws Exception {
		if (connection == null) {
			if (!aconfig.isValid())
				throw new Exception("连接池配置异常");
			// conf = aconfig;// ActConfig.loadFile("mq/activemq.conf");
			// 创建一个链接工厂
			ConnectionFactory connectionFactory = new ActiveMQConnectionFactory(aconfig.getUrl());
			// 从工厂中创建一个链接
			connection = connectionFactory.createConnection(aconfig.getUser(), aconfig.getPassword());

			// 开启链接
			try {
				connection.start();
			} catch (JMSException jmse) {
				connection.close();
				throw jmse;
			}
		}
	}

	private MyActiveMq(ActConfig aconfig) {
		try {
			connect(aconfig);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public static MyActiveMq getInstance() {
		try {
			return getInstance(ActConfig.loadFile(null), TOPIC_DEFAULT, QUEUE_MODEL, Session.AUTO_ACKNOWLEDGE);
		} catch (JMSException e) {
			e.printStackTrace();
			return null;
		}
	}

	/**
	 * 实例化后先初始运行必要的参数
	 * 
	 * @param destination     目标 是 topicname 或queuename
	 * @param model           等于 QUEUE_MODEL表示串行模式,等于TOPIC_MODEL表示订阅模式<br>
	 *                        串行模式QUEUE: 比如发布了10个消息，两个接收者A,B那就是A,B总共会收到10条消息，不重复。
	 *                        订阅模式Topic:一个发布者发布消息，有两个接收者A,B来订阅，那么发布了10条消息，A,B各收到10条消息。
	 * @param acknowledgeMode 表示自动应答还是手动事物应答
	 * @throws JMSException
	 */
	public static MyActiveMq getInstance(ActConfig aconfig, String groupname, int model, Integer acknowledgeMode)
			throws JMSException {
		MyActiveMq mam = new MyActiveMq(aconfig);
		mam.ini(groupname, model, acknowledgeMode);
		return mam;
	}

	/**
	 * 实例化后先初始运行必要的参数，Activemq支持一下三种模式：<br>
	 * * JMS规范的ack消息确认机制有一下四种，定于在session对象中：<br>
	 * AUTO_ACKNOWLEDGE = 1、 自动应答：CLIENT_ACKNOWLEDGE=2客户端手动应答<br>
	 * Session设置为AUTO_ACKNOWLEDGE ，接收到消息（receive 或 onMessage成功返回时），即为消费成功，
	 * 然后从队列里移除该数据。<br>
	 * 不关心该数据有没有正确被处理成我们想要的结果 Session设置为CLIENT_ACKNOWLEDGE 时，必须手动调用acknowledge
	 * 方法才为消费成功 然后从队列里移除该条数据。<br>
	 * onmessage 回调函数中 message.acknowledge(); 执行为应答<br>
	 * DUPS_OK_ACKNOWLEDGE = 3： 自动批量确认,或被多次消费<br>
	 * 在第二次重新传递消息的时候，消息头的JmsDelivered会被置为true标示当前消息已经传送过一次，客户端需要进行消息的重复处理控制。<br>
	 * SESSION_TRANSACTED = 0：事务提交并确认，如果不进行session.commit() 不算确认。 <br>
	 * 可以回滚 此时 message.acknowledge();无效<br>
	 * 如果回滚 _session.rollback(); 那么消息会回滚到队列中，<br>
	 * activemq补充了一个自定义的ACK模式， INDIVIDUAL_ACKNOWLEDGE = 4：单条消息确认<br>
	 * 
	 * @param destination     目标 是 topicname 或queuename
	 * @param model           等于 QUEUE_MODEL表示串行模式,等于TOPIC_MODEL表示订阅模式<br>
	 *                        串行模式QUEUE:
	 *                        比如发布了10个消息，两个接收者A,B那就是A,B总共会收到10条消息，不重复。<br>
	 *                        订阅模式Topic:一个发布者发布消息，有两个接收者A,B来订阅，那么发布了10条消息，A,B各收到10条消息。<br>
	 * @param acknowledgeMode 表示自动应答还是手动事物应答,
	 * 
	 * @throws JMSException
	 */
	private void ini(String group, int model, Integer acknowledgeMode) throws JMSException {
		if (_session != null)
			return;

		if (acknowledgeMode != null && acknowledgeMode >= 0 && acknowledgeMode < 5)
			_acknowledgeMode = acknowledgeMode;
		boolean transacted = true;
		if (_acknowledgeMode > 0)
			transacted = false;
		// System.out.println("_acknowledgeMode:" + _acknowledgeMode + "," +
		// transacted);
		_session = connection.createSession(transacted, _acknowledgeMode);
		if (_destination == null) {
			if (model == QUEUE_MODEL)
				_destination = _session.createQueue(group);
			else
				_destination = _session.createTopic(group);
			// System.out.println("_acknowledgeMode:" + _acknowledgeMode + "," + transacted
			// + model + group);
		}
	}

	/** 一个会话可以创建多个生产者 */
	private MessageProducer getProducer() throws JMSException {
		// 创建一个消息队列
		// 消费者
		// MessageConsumer _consumer = null;
		// 消息生产者
		// MessageProducer producer = LocalProducer.get();
		if (producer == null) {
			producer = _session.createProducer(_destination);
			// 发送模式 DeliveryMode.NON_PERSISTENT=1:非持久 ; DeliveryMode.PERSISTENT=2:持久
			producer.setDeliveryMode(DeliveryMode.PERSISTENT); // 是否为持久化 ，NON_PERSISTENT非持久化 ，PERSISTENT持久化
			// LocalProducer.set(producer);
		}
		return producer;
	}

	private void send(Message msg) throws JMSException {
		getProducer().send(msg);
	}

	public void sendText(String msg) throws JMSException {
		TextMessage test = _session.createTextMessage(msg);
		send(test);
	}

	public void sendBytes(byte[] msg) throws JMSException {
		BytesMessage test = _session.createBytesMessage();
		test.writeBytes(msg);
		send(test);
	}

	public void sendMap(Map<String, String> msg) throws JMSException {
		// TextMessage msg = _session
		// .createTextMessage("productor:我是大帅哥，我现在正在生产东西！");
		// 发送消息
		MapMessage test = _session.createMapMessage();
		msg.forEach((key, value) -> {
			try {
				test.setString(key, value);
			} catch (JMSException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		});
		send(test);
	}

	private MessageConsumer getConsumer() throws JMSException {
		// MessageConsumer consumer = LocalConsumer.get();
		// System.out.println("consumer1"+consumer);
		if (consumer == null) {
			// System.out.println("consumer11"+_destination.toString());
			consumer = _session.createConsumer(_destination);
			// consumer.set(DeliveryMode.NON_PERSISTENT); //是否为持久化 ，NON_PERSISTENT非持久化
			// ，PERSISTENT持久化
			// LocalConsumer.set(consumer);
		}
		// System.out.println("consumer22"+consumer);
		return consumer;
	}

	/**
	 * 如果调用 receiveText() receiveMap() receiveBytes() 就立即应答 如果要手动应答必须调用回调函数
	 */
	public void acknowledge(Message message) {
		if (_acknowledgeMode > 0)
			try {
				message.acknowledge();
			} catch (JMSException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
	}

	// 不确定的情况
	/**
	 * 最原始的形态方法,没有经过封装,需要直接转化消息格式
	 */
	public Message receive() throws JMSException {
		return getConsumer().receive();
	}

	/**
	 * 最原始的形态方法,没有经过封装,需要直接转化消息格式
	 */
	public void setMessageListener(MessageListener listener) throws JMSException {
		getConsumer().setMessageListener(listener);
	}

	/**
	 * 改造的方法,已经可以接收文本,map,bytes
	 * 
	 * @param done
	 * @throws Exception
	 */
	public void receive(IReceive done) throws Exception {
		Message message = getConsumer().receive();

		Object value = null;
		if (message instanceof TextMessage) {
			// System.out.println("receiveText2");
			value = ((TextMessage) message).getText();
		} else if (message instanceof MapMessage) {
			value = getMap(message);
		} else if (message instanceof BytesMessage) {
			value = getBytes(message);
		}
		done.run(value, message);
	}

	// 可以紧接着 应答手动调用message.acknowledge();
	/**
	 * 傻瓜获取文本内容的方法,接收后 会自动应答, 所以你设置 Session.CLIENT_ACKNOWLEDGE 是无效的
	 * 如果不想用自动应答可以用receive
	 * 
	 * @return
	 * @throws JMSException
	 */
	public String receiveText() throws JMSException {
		// System.out.println("receiveText0");
		Message message = getConsumer().receive();
		acknowledge(message);
		if (message instanceof TextMessage) {
			// System.out.println("receiveText2");
			return ((TextMessage) message).getText();
		}
		return null;
	}

	@SuppressWarnings("unchecked")
	Map<String, String> getMap(Message message) throws JMSException {
		MapMessage mmsg = (MapMessage) message;
		Enumeration<String> e = mmsg.getMapNames();
		Map<String, String> mmap = new HashMap<String, String>();
		while (e.hasMoreElements()) {
			String key = e.nextElement();
			mmap.put(key, mmsg.getString(key));
		}
		return mmap;
	}

	/**
	 * 傻瓜获取Map键值对的方法 ,接收后会自动应答,如果不想用自动应答可以用receive 所以你设置 Session.CLIENT_ACKNOWLEDGE
	 * 是无效的
	 * 
	 * @return
	 * @throws JMSException
	 */
	public Map<String, String> receiveMap() throws JMSException {
		Message message = getConsumer().receive();
		acknowledge(message);
		if (message instanceof MapMessage) {
			return getMap(message);
		}
		return null;
	}

	byte[] getBytes(Message message) throws JMSException {
		BytesMessage bm = (BytesMessage) message;
		int len = (int) bm.getBodyLength();
		byte[] bytes = new byte[len];
		bm.readBytes(bytes);
		return bytes;
	}

	/**
	 * 傻瓜获取bytes的方法,接收后 会自动应答,如果不想用自动应答可以用receive
	 * 
	 * @return
	 * @throws JMSException
	 */
	public byte[] receiveBytes() throws JMSException {
		Message message = getConsumer().receive();
		acknowledge(message);
		if (message instanceof BytesMessage) {
			return getBytes(message);
		}

		return null;
	}

	public void setMessageListener(IDo<Message> done) throws JMSException {
		// 创建一个消息队列
		getConsumer().setMessageListener(new MessageListener() {
			public void onMessage(Message message) {
				try {
					done.run(message);
				} catch (Exception e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		});
	}

	public void rollback() throws JMSException {
		if (_acknowledgeMode == Session.SESSION_TRANSACTED)
			_session.rollback();
	}

	public void commit() {
		if (_acknowledgeMode == Session.SESSION_TRANSACTED)
			try {
				_session.commit();
			} catch (JMSException e) {
				try {
					_session.rollback();
				} catch (JMSException e1) {
					// TODO Auto-generated catch block
					e1.printStackTrace();
				}
			}
	}

	public void close() {
		try {
			_session.close();
			// MyDate.sleep(1000);
			connection.close();
		} catch (Exception e) {

		}
	}
}
